// Copyright (c) 2018 Uber Technologies, Inc.
//
// Permission is hereby granted, free of charge, to any person obtaining a copy
// of this software and associated documentation files (the "Software"), to deal
// in the Software without restriction, including without limitation the rights
// to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
// copies of the Software, and to permit persons to whom the Software is
// furnished to do so, subject to the following conditions:
//
// The above copyright notice and this permission notice shall be included in
// all copies or substantial portions of the Software.
//
// THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
// IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
// FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
// AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
// LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
// OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
// THE SOFTWARE.

package topic

import (
	"bytes"
	"errors"
	"fmt"
	"time"

	"github.com/m3db/m3/src/cluster/kv"
	"github.com/m3db/m3/src/cluster/services"
	"github.com/m3db/m3/src/msg/generated/proto/topicpb"
)

var (
	errEmptyName  = errors.New("invalid topic: empty name")
	errZeroShards = errors.New("invalid topic: zero shards")
)

type topic struct {
	name             string
	numOfShards      uint32
	consumerServices []ConsumerService
	version          int
}

// NewTopic creates a new topic.
func NewTopic() Topic { return new(topic) }

// NewTopicFromValue creates a topic from a kv.Value.
func NewTopicFromValue(v kv.Value) (Topic, error) {
	var topic topicpb.Topic
	if err := v.Unmarshal(&topic); err != nil {
		return nil, err
	}
	t, err := NewTopicFromProto(&topic)
	if err != nil {
		return nil, err
	}
	return t.SetVersion(v.Version()), nil
}

// NewTopicFromProto creates a topic from a proto.
func NewTopicFromProto(t *topicpb.Topic) (Topic, error) {
	css := make([]ConsumerService, len(t.ConsumerServices))
	for i, cspb := range t.ConsumerServices {
		cs, err := NewConsumerServiceFromProto(cspb)
		if err != nil {
			return nil, err
		}
		css[i] = cs
	}
	return NewTopic().
		SetName(t.Name).
		SetNumberOfShards(t.NumberOfShards).
		SetConsumerServices(css), nil
}

func (t *topic) Name() string {
	return t.name
}

func (t *topic) SetName(value string) Topic {
	newt := *t
	newt.name = value
	return &newt
}

func (t *topic) NumberOfShards() uint32 {
	return t.numOfShards
}

func (t *topic) SetNumberOfShards(value uint32) Topic {
	newt := *t
	newt.numOfShards = value
	return &newt
}

func (t *topic) ConsumerServices() []ConsumerService {
	return t.consumerServices
}

func (t *topic) SetConsumerServices(value []ConsumerService) Topic {
	newt := *t
	newt.consumerServices = value
	return &newt
}

func (t *topic) Version() int {
	return t.version
}

func (t *topic) SetVersion(value int) Topic {
	newt := *t
	newt.version = value
	return &newt
}

func (t *topic) AddConsumerService(value ConsumerService) (Topic, error) {
	cur := t.ConsumerServices()
	for _, cs := range cur {
		if cs.ServiceID().Equal(value.ServiceID()) {
			return nil, fmt.Errorf("service %s is already consuming the topic", value.ServiceID().String())
		}
	}

	return t.SetConsumerServices(append(cur, value)), nil
}

func (t *topic) RemoveConsumerService(value services.ServiceID) (Topic, error) {
	cur := t.ConsumerServices()
	for i, cs := range cur {
		if cs.ServiceID().Equal(value) {
			cur = append(cur[:i], cur[i+1:]...)
			return t.SetConsumerServices(cur), nil
		}
	}

	return nil, fmt.Errorf("could not find consumer service %s in the topic", value.String())
}

func (t *topic) UpdateConsumerService(value ConsumerService) (Topic, error) {
	css := t.ConsumerServices()
	for i, cs := range css {
		if !cs.ServiceID().Equal(value.ServiceID()) {
			continue
		}
		if value.ConsumptionType() != cs.ConsumptionType() {
			return nil, fmt.Errorf("could not change consumption type for consumer service %s", value.ServiceID().String())
		}
		css[i] = value
		return t.SetConsumerServices(css), nil
	}
	return nil, fmt.Errorf("could not find consumer service %s in the topic", value.String())
}

func (t *topic) String() string {
	var buf bytes.Buffer
	buf.WriteString("\n{\n")
	buf.WriteString(fmt.Sprintf("\tversion: %d\n", t.version))
	buf.WriteString(fmt.Sprintf("\tname: %s\n", t.name))
	buf.WriteString(fmt.Sprintf("\tnumOfShards: %d\n", t.numOfShards))
	if len(t.consumerServices) > 0 {
		buf.WriteString("\tconsumerServices: {\n")
	}
	for _, cs := range t.consumerServices {
		buf.WriteString(fmt.Sprintf("\t\t%s\n", cs.String()))
	}
	if len(t.consumerServices) > 0 {
		buf.WriteString("\t}\n")
	}
	buf.WriteString("}\n")
	return buf.String()
}

func (t *topic) Validate() error {
	if t.Name() == "" {
		return errEmptyName
	}
	if t.NumberOfShards() == 0 {
		return errZeroShards
	}
	uniqConsumers := make(map[string]struct{}, len(t.ConsumerServices()))
	for _, cs := range t.ConsumerServices() {
		_, ok := uniqConsumers[cs.ServiceID().String()]
		if ok {
			return fmt.Errorf("invalid topic: duplicated consumer %s", cs.ServiceID().String())
		}
		uniqConsumers[cs.ServiceID().String()] = struct{}{}

		filterConfig := cs.DynamicFilterConfigs()

		if filterConfig != nil {
			shardSetFilter := filterConfig.ShardSetFilter()
			if shardSetFilter != nil {
				shardSet := shardSetFilter.ShardSet()
				if shardSet == "" {
					return fmt.Errorf("invalid topic: empty shard set in filter for consumer %s", cs.ServiceID().String())
				}
			}

			percentageFilter := filterConfig.PercentageFilter()
			if percentageFilter != nil {
				percentage := percentageFilter.Percentage()
				if percentage <= 0 || percentage >= 100 {
					return fmt.Errorf("invalid topic: invalid percentage in filter for consumer %s", cs.ServiceID().String())
				}
			}

			storagePolicyFilter := filterConfig.StoragePolicyFilter()
			if storagePolicyFilter != nil {
				storagePolicies := storagePolicyFilter.StoragePolicies()
				if len(storagePolicies) == 0 {
					return fmt.Errorf("invalid topic: empty storage policy filter for consumer %s", cs.ServiceID().String())
				}

				for _, storagePolicy := range storagePolicies {
					if storagePolicy == "" {
						return fmt.Errorf("invalid topic: empty storage policy in filter for consumer %s", cs.ServiceID().String())
					}
				}
			}
		}
	}

	return nil
}

// ToProto creates proto from a topic.
func ToProto(t Topic) (*topicpb.Topic, error) {
	css := t.ConsumerServices()
	csspb := make([]*topicpb.ConsumerService, len(css))
	for i, cs := range css {
		cspb, err := ConsumerServiceToProto(cs)
		if err != nil {
			return nil, err
		}
		csspb[i] = cspb
	}
	return &topicpb.Topic{
		Name:             t.Name(),
		NumberOfShards:   t.NumberOfShards(),
		ConsumerServices: csspb,
	}, nil
}

type filterConfig struct {
	shardSetFilterConfig      *shardSetFilter
	percentageFilterConfig    *percentageFilter
	storagePolicyFilterConfig *storagePolicyFilter
	routingPolicyFilterConfig *routingPolicyFilter
}

// NewFilterConfig creates a new filter config.
func NewFilterConfig() FilterConfig {
	return new(filterConfig)
}

func (fc *filterConfig) ShardSetFilter() ShardSetFilter {
	if fc.shardSetFilterConfig == nil {
		return nil
	}

	return fc.shardSetFilterConfig
}

func (fc *filterConfig) SetShardSetFilter(value ShardSetFilter) FilterConfig {
	newfc := *fc
	if value != nil {
		newfc.shardSetFilterConfig = value.(*shardSetFilter)
	}
	return &newfc
}

func (fc *filterConfig) PercentageFilter() PercentageFilter {
	if fc.percentageFilterConfig == nil {
		return nil
	}

	return fc.percentageFilterConfig
}

func (fc *filterConfig) SetPercentageFilter(value PercentageFilter) FilterConfig {
	newfc := *fc
	if value != nil {
		newfc.percentageFilterConfig = value.(*percentageFilter)
	}
	return &newfc
}

func (fc *filterConfig) StoragePolicyFilter() StoragePolicyFilter {
	if fc.storagePolicyFilterConfig == nil {
		return nil
	}

	return fc.storagePolicyFilterConfig
}

func (fc *filterConfig) SetStoragePolicyFilter(value StoragePolicyFilter) FilterConfig {
	newfc := *fc
	if value != nil {
		newfc.storagePolicyFilterConfig = value.(*storagePolicyFilter)
	}
	return &newfc
}

func (fc *filterConfig) RoutingPolicyFilter() RoutingPolicyFilter {
	if fc.routingPolicyFilterConfig == nil {
		return nil
	}

	return fc.routingPolicyFilterConfig
}

func (fc *filterConfig) SetRoutingPolicyFilter(value RoutingPolicyFilter) FilterConfig {
	newfc := *fc
	if value != nil {
		newfc.routingPolicyFilterConfig = value.(*routingPolicyFilter)
	}
	return &newfc
}

type shardSetFilter struct {
	shardSet string
}

func (filter *shardSetFilter) ShardSet() string {
	return filter.shardSet
}

// NewShardSetFilter creates a new shard set filter.
func NewShardSetFilter(shardSet string) ShardSetFilter {
	return &shardSetFilter{shardSet: shardSet}
}

type percentageFilter struct {
	percentage float64
}

// NewPercentageFilter creates a new percentage filter.
func NewPercentageFilter(percentage float64) PercentageFilter {
	return &percentageFilter{percentage: percentage}
}

func (filter *percentageFilter) Percentage() float64 {
	return filter.percentage
}

type storagePolicyFilter struct {
	storagePolicies []string
}

// NewStoragePolicyFilter creates a new storage policy filter.
func NewStoragePolicyFilter(storagePolicies []string) StoragePolicyFilter {
	return &storagePolicyFilter{storagePolicies: storagePolicies}
}

func (filter *storagePolicyFilter) StoragePolicies() []string {
	return filter.storagePolicies
}

type routingPolicyFilter struct {
	allowedTrafficTypes []string
	isDefault           bool
}

// AllowedTrafficTypes returns the traffic types that the filter will return true for.
func (filter *routingPolicyFilter) AllowedTrafficTypes() []string {
	return filter.allowedTrafficTypes
}

// IsDefault determines the behavior of the filter when no traffic type is
// specified. If true, the filter will return true.
func (filter *routingPolicyFilter) IsDefault() bool {
	return filter.isDefault
}

type consumerService struct {
	sid           services.ServiceID
	ct            ConsumptionType
	ttlNanos      int64
	filterConfigs *filterConfig
}

// NewConsumerService creates a ConsumerService.
func NewConsumerService() ConsumerService {
	return new(consumerService)
}

// NewConsumerServiceFromProto creates a ConsumerService from a proto.
func NewConsumerServiceFromProto(cs *topicpb.ConsumerService) (ConsumerService, error) {
	ct, err := NewConsumptionTypeFromProto(cs.ConsumptionType)
	if err != nil {
		return nil, err
	}
	return NewConsumerService().
		SetServiceID(NewServiceIDFromProto(cs.ServiceId)).
		SetConsumptionType(ct).
		SetMessageTTLNanos(cs.MessageTtlNanos).
		SetDynamicFilterConfigs(NewDynamicFilterConfigFromProto(cs.Filters)), nil
}

// ConsumerServiceToProto creates proto from a ConsumerService.
func ConsumerServiceToProto(cs ConsumerService) (*topicpb.ConsumerService, error) {
	ct, err := ConsumptionTypeToProto(cs.ConsumptionType())
	if err != nil {
		return nil, err
	}
	return &topicpb.ConsumerService{
		ConsumptionType: ct,
		ServiceId:       ServiceIDToProto(cs.ServiceID()),
		MessageTtlNanos: cs.MessageTTLNanos(),
		Filters:         DynamicFilterConfigToProto(cs.DynamicFilterConfigs()),
	}, nil
}

func (cs *consumerService) ServiceID() services.ServiceID {
	return cs.sid
}

func (cs *consumerService) SetServiceID(value services.ServiceID) ConsumerService {
	newcs := *cs
	newcs.sid = value
	return &newcs
}

func (cs *consumerService) ConsumptionType() ConsumptionType {
	return cs.ct
}

func (cs *consumerService) SetConsumptionType(value ConsumptionType) ConsumerService {
	newcs := *cs
	newcs.ct = value
	return &newcs
}

func (cs *consumerService) MessageTTLNanos() int64 {
	return cs.ttlNanos
}

func (cs *consumerService) SetMessageTTLNanos(value int64) ConsumerService {
	newcs := *cs
	newcs.ttlNanos = value
	return &newcs
}

func (cs *consumerService) DynamicFilterConfigs() FilterConfig {
	if cs.filterConfigs == nil {
		return nil
	}

	return cs.filterConfigs
}

func (cs *consumerService) SetDynamicFilterConfigs(value FilterConfig) ConsumerService {
	newcs := *cs
	if value != nil {
		newcs.filterConfigs = value.(*filterConfig)
	} else {
		newcs.filterConfigs = nil
	}
	return &newcs
}

func (cs *consumerService) String() string {
	var buf bytes.Buffer
	buf.WriteString("{")
	buf.WriteString(fmt.Sprintf("service: %s, consumption type: %s", cs.sid.String(), cs.ct.String()))
	if cs.ttlNanos != 0 {
		buf.WriteString(fmt.Sprintf(", ttl: %v", time.Duration(cs.ttlNanos)))
	}
	if cs.filterConfigs != nil {
		if cs.filterConfigs.shardSetFilterConfig != nil {
			buf.WriteString(fmt.Sprintf(", shard set filter: %s", cs.filterConfigs.shardSetFilterConfig.shardSet))
		}
		if cs.filterConfigs.percentageFilterConfig != nil {
			buf.WriteString(fmt.Sprintf(", percentage filter: %v", cs.filterConfigs.percentageFilterConfig.percentage))
		}
		if cs.filterConfigs.storagePolicyFilterConfig != nil {
			buf.WriteString(
				fmt.Sprintf(", storage policy filter: %v", cs.filterConfigs.storagePolicyFilterConfig.storagePolicies))
		}
		if cs.filterConfigs.routingPolicyFilterConfig != nil {
			buf.WriteString(fmt.Sprintf(", routing policy filter allowed traffic types: %v",
				cs.filterConfigs.routingPolicyFilterConfig.allowedTrafficTypes))
			buf.WriteString(fmt.Sprintf(", routing policy filter is default: %v",
				cs.filterConfigs.routingPolicyFilterConfig.isDefault))
		}
	}
	buf.WriteString("}")
	return buf.String()
}

// NewServiceIDFromProto creates service id from a proto.
func NewServiceIDFromProto(sid *topicpb.ServiceID) services.ServiceID {
	return services.NewServiceID().SetName(sid.Name).SetEnvironment(sid.Environment).SetZone(sid.Zone)
}

// ServiceIDToProto creates proto from a service id.
func ServiceIDToProto(sid services.ServiceID) *topicpb.ServiceID {
	return &topicpb.ServiceID{
		Name:        sid.Name(),
		Environment: sid.Environment(),
		Zone:        sid.Zone(),
	}
}

// NewDynamicFilterConfigFromProto creates filter config from a proto.
func NewDynamicFilterConfigFromProto(filterProto *topicpb.Filters) FilterConfig {
	if filterProto == nil {
		return nil
	}

	filter := filterConfig{}
	if filterProto.ShardSetFilter != nil {
		filter.shardSetFilterConfig = &shardSetFilter{shardSet: filterProto.ShardSetFilter.ShardSet}
	}
	if filterProto.PercentageFilter != nil {
		filter.percentageFilterConfig = &percentageFilter{percentage: filterProto.PercentageFilter.Percentage}
	}
	if filterProto.StoragePolicyFilter != nil {
		filter.storagePolicyFilterConfig = &storagePolicyFilter{
			storagePolicies: filterProto.StoragePolicyFilter.StoragePolicies}
	}
	if filterProto.RoutingPolicyFilter != nil {
		filter.routingPolicyFilterConfig = &routingPolicyFilter{
			allowedTrafficTypes: filterProto.RoutingPolicyFilter.AllowedTrafficTypes,
			isDefault:           filterProto.RoutingPolicyFilter.IsDefault,
		}
	}

	return &filter
}

// DynamicFilterConfigToProto creates proto from a filter config.
func DynamicFilterConfigToProto(filter FilterConfig) *topicpb.Filters {
	if filter == nil {
		return nil
	}

	return &topicpb.Filters{
		ShardSetFilter:      ShardSetFilterToProto(filter.ShardSetFilter()),
		PercentageFilter:    PercentageFilterToProto(filter.PercentageFilter()),
		StoragePolicyFilter: StoragePolicyFilterToProto(filter.StoragePolicyFilter()),
		RoutingPolicyFilter: RoutingPolicyFilterToProto(filter.RoutingPolicyFilter()),
	}
}

// ShardSetFilterToProto creates proto from a shard set filter.
func ShardSetFilterToProto(filter ShardSetFilter) *topicpb.ShardSetFilter {
	if filter == nil {
		return nil
	}

	return &topicpb.ShardSetFilter{ShardSet: filter.ShardSet()}
}

// PercentageFilterToProto creates proto from a percentage filter.
func PercentageFilterToProto(filter PercentageFilter) *topicpb.PercentageFilter {
	if filter == nil {
		return nil
	}

	return &topicpb.PercentageFilter{Percentage: filter.Percentage()}
}

// StoragePolicyFilterToProto creates proto from a storage policy filter.
func StoragePolicyFilterToProto(filter StoragePolicyFilter) *topicpb.StoragePolicyFilter {
	if filter == nil {
		return nil
	}

	return &topicpb.StoragePolicyFilter{StoragePolicies: filter.StoragePolicies()}
}

// RoutingPolicyFilterToProto creates proto from a routing policy filter.
func RoutingPolicyFilterToProto(filter RoutingPolicyFilter) *topicpb.RoutingPolicyFilter {
	if filter == nil {
		return nil
	}

	return &topicpb.RoutingPolicyFilter{AllowedTrafficTypes: filter.AllowedTrafficTypes(), IsDefault: filter.IsDefault()}
}
